package com.timeriver.feature_project

import org.apache.spark.ml.feature.{PCA, PCAModel}
import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object PCAAlg {
  def main(args: Array[String]): Unit = {

    val session: SparkSession = SparkSession.builder()
      .master("local[6]")
      .appName("PCA抽取有效的潜在因子")
      .getOrCreate()

    import session.implicits._

    val value: Dataset[String] = session.read
      .textFile("D:\\workspace\\gitee_space\\spark-ml-machine-learning\\data\\processed.cleveland.data")

    val data: Dataset[Tuple1[linalg.Vector]] = value.filter(line => !(line.isEmpty || line.contains("?")))
      .map(line => {
        val array: Array[Double] = line.split(",").map(_.toDouble)
        Tuple1.apply(Vectors.dense(array))
      })

    val model: PCAModel = new PCA().setInputCol("_1")
      .setOutputCol("pca-features")
      .setK(5)
      .fit(data)

    val frame: DataFrame = model.transform(data).select("pca-features")

    frame.show(5, false)

    session.stop()
  }
}
